package com.lc1993929.myhdfs.client;

import java.io.FileInputStream;
import java.io.IOException;
import java.io.InputStream;
import java.io.OutputStream;
import java.net.Socket;

/**
 * Created by LiuChang on 2017/9/1/001.
 */
public class Client {
    public static void main(String[] args) throws Exception {
        //跟集群中的NameNode通信，请求上传文件快，NameNode应该相应一个blockID，以及可用的datanode主机

        String response = callNN(1);

        String[] split = response.split(":");
        String blkId = split[0];
        String dnHost = split[1];

        System.out.println(blkId);
        System.out.println(dnHost);

        //连接datanode
        Socket toDN = new Socket("localhost", 9900);
        //先告知DN，即将传输一个BLOCK，并告知全局ID
        OutputStream toDNout = toDN.getOutputStream();
        String toDnRequest = "PUTBLOCK:" + blkId;
        toDNout.write(toDnRequest.getBytes());
        toDNout.flush();

        FileInputStream fin = new FileInputStream("f:/a.log");
        byte[] buf = new byte[4096];
        int len = 0;
        int count = 0;
        while ((len = fin.read(buf)) != -1) {
            count += len;
            toDNout.write(buf, 0, len);
            toDNout.flush();
            if (count == 1 * 1024 * 1024) {
                toDNout.close();
                toDN.close();
                break;
            }
        }


        //块1传完之后，再请求nn获取第二块的blockid及dn
        response = callNN(2);
        split = response.split(":");
        blkId = split[0];
        dnHost = split[1];

        //再请求新的dn，传输文件块

        //连接datanode
        toDN = new Socket("localhost", 9900);
        //先告知DN，即将传输一个BLOCK，并告知全局ID
        toDNout = toDN.getOutputStream();
        toDnRequest = "PUTBLOCK:" + blkId;
        toDNout.write(toDnRequest.getBytes());
        toDNout.flush();

        buf = new byte[4096];
        len = 0;
        count = 0;
        while ((len = fin.read(buf)) != -1) {
            count += len;
            toDNout.write(buf, 0, len);
            toDNout.flush();
            if (count == 1 * 1024 * 1024) {
                toDNout.close();
                toDN.close();
                break;
            }
        }
    }

    private static String callNN(int num) throws IOException {
        Socket toNN = new Socket("localhost", 9000);
        OutputStream toNNout = toNN.getOutputStream();
        InputStream toNNin = toNN.getInputStream();

        toNNout.write(("PUTBLOCK:/aa/a.log:" + num).getBytes());
        toNNout.flush();

        byte[] b = new byte[1024];
        int read = toNNin.read(b);

        //规定：服务端返回的数据格式：BLK_001:hdp-dn-01
        //假设datanode的工作端口都为9900
        String response = new String(b, 0, read);


        toNNin.close();
        toNNout.close();
        toNN.close();
        return response;
    }
}
